package com.icodingedu.springcloud.service;

import com.icodingedu.springcloud.entity.MessageBean;
import com.icodingedu.springcloud.topic.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.cloud.stream.messaging.Sink;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.messaging.Message;
import org.springframework.messaging.handler.annotation.Header;

import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@EnableBinding(value = {
        Sink.class,
        MyTopic.class,
        GroupTopic.class,
        DelayedTopic.class,
        ErrorTopic.class,
        RequeueTopic.class,
        DlqTopic.class,
        FallbackTopic.class
})
public class StreamConsumer {

    private AtomicInteger count = new AtomicInteger(1);

    //这里使用的是一个默认的topic
    @StreamListener(Sink.INPUT)
    public void consumer(Object payload){
        log.info("message consumed successfully,payload={}",payload);
    }

    @StreamListener(MyTopic.INPUT)
    public void consumerMyMessage(Object payload){
        log.info("My message consumed successfully,payload={}",payload);
    }

    @StreamListener(GroupTopic.INPUT)
    public void consumerGroupMessage(Object payload){
        log.info("Group message consumed successfully,payload={}",payload);
    }

    @StreamListener(DelayedTopic.INPUT)
    public void consumerDmMessage(MessageBean messageBean){
        log.info("Group message consumed successfully,payload={}",messageBean.getPayload());
    }

    //异常重试单机版
    @StreamListener(ErrorTopic.INPUT)
    public void consumerErrorMessage(Object payload){
        log.info("***** 进入异常处理 *****");
        if(count.incrementAndGet() %3 == 0){
            log.info("====== 完全没有问题! =====");
            count.set(0);
        }else{
            log.info("--------- what's your problem? ---------");
            throw new RuntimeException("********** 整个人都不行了 **********");
        }
    }

    @StreamListener(RequeueTopic.INPUT)
    public void consumerRQMessage(Object payload){
        log.info("***** 进入入队异常处理 *****");
        try{
            Thread.sleep(3000);
        }catch (Exception ex){
            log.error("***延迟等待错误{}***",ex);
        }
        //让消息一只重回队列
        throw new RuntimeException("********** 整个人都不行了 **********");
    }

    //死信队列
    @StreamListener(DlqTopic.INPUT)
    public void consumerDLQMessage(Object payload){
        log.info("***** DLQ 进入异常处理 *****");
        if(count.incrementAndGet() %3 == 0){
            log.info("====== DLK 完全没有问题! =====");
        }else{
            log.info("--------- what's your problem? ---------");
            throw new RuntimeException("********** DLK 整个人都不行了 **********");
        }
    }

    @StreamListener(FallbackTopic.INPUT)
    public void consumerFallbackMessage(Object payload, @Header("version") String version){
        log.info("***** Fallback Are you Ok? *****");
        if("1.0".equalsIgnoreCase(version)){
            log.info("====== Fallback完全没有问题 ======");
        }else if("2.0".equalsIgnoreCase(version)){
            log.info("====== unsupported version ======");
            throw new RuntimeException("******* fallback version *******");
        }else{
            log.info("------ Fallback Version={} ------",version);
        }
    }
    //exchange.group.errors
    //topic配置的时候一定要配置组名
    @ServiceActivator(inputChannel = "fallback-exchange.Fallback-Group.errors")
    public void fallback(Message<?> message){
        log.info("***** Enter Fallback ****");
    }
}
